package com.gzsxy.rabbitmq.consumer.receiver;

import com.gzsxy.rabbitmq.consumer.config.redis.StringRedisTemplate;
import com.gzsxy.rabbitmq.consumer.entity.MessageIdempotent;
import com.gzsxy.rabbitmq.consumer.service.MessageService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.EnableAspectJAutoProxy;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 消费端-手动确认ack，
 * 需要配置 spring.rabbitmq.listener.acknowledge-mode=manual  消费者开启手动ack消息确认
 * 需要配置 spring.rabbitmq.listener.default-requeue-rejected=false  设置为false，会重发消息到死信队列（采用自动重试才需要配置）
 * 消费者端防止消息丢失，方案一：
 * 1.消费者开始手动ack消息确定
 * 2.尝试消息重投
 * 3.对重投还是消费失败的消息拒绝确认
 * 4.把拒绝确认的消息扔到死信队列中，并记录到数据库，再做好消息预警
 * </p>
 * @Author: xiaolong
 * @date: 2023/2/16 10:05
 * @Version: 1.0
 */
@Slf4j
@Component
@EnableAspectJAutoProxy(proxyTargetClass = true,exposeProxy = true)  //AopContext.currentProxy()需要
@ConditionalOnProperty(value = "rabbitmq.enable.type",havingValue = "reliable",matchIfMissing = false)
public class AckConsumer {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private MessageService messageService;

    /**
     * 消息监听方法
     * bindings: 完成队列与交换机的绑定
     * Queue: 队列属性，超过最大值，超时未被消费，消费失败超过重试次数，都会被仍到信息队列中
     * exchange：交换机属性
     * key：路由key,通配符
     */
    @RabbitListener(queues = "QE")
    public void handlerMessage(String msg, Channel channel, Message message)  {
        try {
            String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");

            log.info("【队列QE】接受到消息msg=【{}】,准备消费", msg);
            //0. 获取MessageId, 消息唯一id
            log.info("获取到消息id=【】",messageId);
            //1. 设置key到Redis  解决消息幂等性问题 (redis过期时间会存在key过期后，rabbitmq才重启动，重复消费)
            Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(messageId, "0");

//            Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(messageId, "0", 60, TimeUnit.SECONDS);
            if (aBoolean) {
//                String msg1 = new String(message.getBody());
                log.info("当前时间：{},收到QB队列信息{}", new Date().toString(), msg);
                //2. 消费消息
                //业务操作 AopContext.currentProxy()防止事务失效，使用spirngAOP进行管理
                ((AckConsumer) AopContext.currentProxy()).addErrorMsg(messageId);
                // 手动ack确认
                //参数1：deliveryTag:消息唯一传输ID
                //参数2：multiple：true: 手动批量处理，false: 手动单条处理
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

            } else {
                //5. 获取Redis中的value即可 如果是1，手动ack
                if ("1".equalsIgnoreCase(String.valueOf(stringRedisTemplate.opsForValue().get(messageId)))) {
                    // 手动ack确认
                    //参数1：deliveryTag:消息唯一传输ID
                    //参数2：multiple：true: 手动批量处理，false: 手动单条处理
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } else {
                    //若再进行业务操作过程中出现异常但是缓存值没更新，消息重试业务数据回滚需要重新执行，执行成功返回ack，失败投入死信队列
                    ((AckConsumer) AopContext.currentProxy()).addErrorMsg(messageId);
                    // 手动ack确认
                    //参数1：deliveryTag:消息唯一传输ID
                    //参数2：multiple：true: 手动批量处理，false: 手动单条处理
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                }
            }
            log.info("【队列QE】消息msg=【{}】,消费成功", msg);
        } catch (Exception ex) {
            // 如果真得出现了异常，我们采用消息重投,获取redelivered，判断是否为重投: false没有重投，true重投
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            try {
                // (已重投)拒绝确认
                if (redelivered) {
                    /**
                     * 拒绝确认，从队列中删除该消息，防止队列阻塞(消息堆积)
                     * boolean requeue: false不重新入队列(丢弃消息)
                     */
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                    log.info("【队列QE】消费的消息msg=【{}】,已投入死信队列", msg);
                } else { // (没有重投) 消息重投
                    /**
                     * 消息重投，重新把消息放回队列中
                     * boolean multiple: 单条或批量
                     * boolean requeue: true重回队列
                     */
                    try {
                        Thread.sleep(2000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    log.info("【队列QE】消息msg=【{}】重投了",msg);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 消费者 - 死信队列
     * @author
     */
    @RabbitListener(queues = "QF")
    public void receiveF(Message message, Channel channel) throws IOException {
        //0. 获取MessageId, 消息唯一id
        String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        //1. 设置key到Redis
        if (stringRedisTemplate.opsForValue().setIfAbsent(messageId, "0", 10, TimeUnit.SECONDS)) {
            String msg = new String(message.getBody());
            log.info("当前时间：{},收到QB队列信息{}", new Date().toString(), msg);
            //2. 消费消息
            System.out.println("接收到消息：" + msg);
            //业务操作
            addErrorMsg(messageId);
            //3. 设置key的value为1
            stringRedisTemplate.opsForValue().set(messageId, "1", 10, TimeUnit.SECONDS);
            //4.  手动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            //5. 获取Redis中的value即可 如果是1，手动ack
            if ("1".equalsIgnoreCase(String.valueOf(stringRedisTemplate.opsForValue().get(messageId)))) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            }
        }
    }


    //事务Propagation.NESTED模式，内部事务不影响外部事务，如果出现异常内部事务立即会回滚，无需等外部事务
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.NESTED)
    public void addYWCZ(Channel channel,Message message,String messageId) throws IOException {
       //3.设置key的value为1，标识消息在消费中
        stringRedisTemplate.opsForValue().set(messageId,"1");



        // 业务处理
        int i = 10 / 0;
        try {
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        // 手动ack确认
        //参数1：deliveryTag:消息唯一传输ID
        //参数2：multiple：true: 手动批量处理，false: 手动单条处理
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);


        //3. 设置key的value为2 标识消息消费完成
        stringRedisTemplate.opsForValue().set(messageId, "2");

    }


    //事务Propagation.NESTED模式，内部事务不影响外部事务，如果出现异常内部事务立即会回滚，无需等外部事务
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.NESTED)
    public void addErrorMsg(String messageId){
        MessageIdempotent msg = new MessageIdempotent();
        msg.setMessageContent("你好");
        msg.setMessageId(messageId);
        //添加插库操作必须要先查询有无重复数据再进行添加，若有则不进行添加
        messageService.addErrorMsg(msg);
        // 业务处理
        int i = 10 / 0;
        //3. 设置key的value为1 （之所以把redis写进事务,怕redis异常后，）
        stringRedisTemplate.opsForValue().set(messageId, "1", 10, TimeUnit.SECONDS);

    }
}
